package harness;

import edpool.EDPool;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.concurrent.CyclicBarrier;
import java.util.logging.Level;
import java.util.logging.Logger;
import fcqueue.FCQueue;
import java.util.concurrent.atomic.AtomicLong;
import javaqueue.JavaQueueAdapter;
import quasilinqueue.LockFreeQueue;
import rendezvousqueue.RendezvousSyncQueue;
import snpi.SNPIFactoryCASOnce;
import snpi.SNPIQueue;

/**
 *
 * @author trbot
 */
public class Harness {

    private final static boolean DEBUG = false;
    private final static int NANOS_PER_SECOND = 1000000000;
    
    public Harness(PrintStream out) {
        
        // set up parameters for experiments
        
        final int QUEUE_SIZE = 200;
        final int NUMBER_OF_TRIALS = 3;//8;
        
        //final int[] threadCounts = {1,2,3,4,5,6,7,8};
        final int[] threadCounts = {4};//{1,2,4,8,12,16,20,24,28,32,36,40,44,48,52,56,60,64};//,80,96,112,128/*,68,72,76,80,84,88,92,96,100,104,108,112,116,120,124,128*/};
        
        // create mapping to use for binding threads to virtual processors
        final int NUMBER_OF_THREADS = 128;
        final int[] threadProcBinding = new int[NUMBER_OF_THREADS];
        for (int i=0;i<NUMBER_OF_THREADS;i++) threadProcBinding[i] = i;
        
        // create experiments
        
        ArrayList<Trial> trials = new ArrayList<Trial>();
        
        for (int prefill=0;prefill<=0;++prefill) {//1;++prefill) {
//            for (int i=0;i<threadCounts.length;i++) {
//                for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
//                    Queue q = new LockFreeQueue<Long>();
//                    trials.add(new Trial(new RandomWorkThreadFactory(),
//                                            (prefill>0?true:false), QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
//                                            (int) Math.round(Math.random()*Integer.MAX_VALUE),
//                                            q, "QuasiLinQueue"));
//                }
//            }
            for (int i=0;i<threadCounts.length;i++) {
                for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
                    Queue q = new SNPIQueue<Long>(new SNPIFactoryCASOnce(), QUEUE_SIZE);
                    trials.add(new Trial(new RandomWorkThreadFactory(),
                                            (prefill>0?true:false), QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
                                            (int) Math.round(Math.random()*Integer.MAX_VALUE),
                                            q, "SNPIQueue"));
                }
            }
            for (int i=0;i<threadCounts.length;i++) {
                for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
                    Queue q = new JavaQueueAdapter<Long>(QUEUE_SIZE);
                    trials.add(new Trial(new RandomWorkThreadFactory(),
                                            (prefill>0?true:false), QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
                                            (int) Math.round(Math.random()*Integer.MAX_VALUE),
                                            q, "JavaQueue"));
                }
            }
//            for (int i=0;i<threadCounts.length;i++) {
//                for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
//                    Queue q = new FCQueue();
//                    trials.add(new Trial(new RandomWorkThreadFactory(),
//                                            (prefill>0?true:false), QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
//                                            (int) Math.round(Math.random()*Integer.MAX_VALUE),
//                                            q, "FCQueue"));
//                }
//            }
        }
//        
//        for (int i=0;i<threadCounts.length;i++) {
//            for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
//                Queue q = new RendezvousSyncQueue();
//                trials.add(new Trial(new RandomWorkThreadFactory(),
//                                        false, QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
//                                        (int) Math.round(Math.random()*Integer.MAX_VALUE),
//                                        q, "RendezvousQueue"));
//            }
//        }
//        for (int i=0;i<threadCounts.length;i++) {
//            for (int trialno=0;trialno<NUMBER_OF_TRIALS;trialno++) {
//                Queue q = new EDPool<Long>(8, 8, 4);
//                trials.add(new Trial(new RandomWorkThreadFactory(),
//                                        false, QUEUE_SIZE, trialno, 1.0, threadCounts[i], threadProcBinding,
//                                        (int) Math.round(Math.random()*Integer.MAX_VALUE),
//                                        q, "EDPool"));
//            }
//        }



        // run experiments and produce output
        
        out.println("qname,capacity,prefill,nthreads,nseconds,trialno,seed,elapsedSeconds,trueEnqueue,trueDequeue,throughput");
        
        int cnt = 0;
        for (Trial trial : trials) {
            
            // run the experiment
            
            /*if (DEBUG) {*/
                System.err.println("running trial " + (++cnt) + "/" + trials.size() + "; " + trial.qName + ", " + trial.nthreads + " threads, trialno " + trial.trialno);
            // }

            run(trial);
            
            // produce output for the experiment 
            
            out.print(trial.qName);
            out.print("," + trial.queueCapacity);
            out.print("," + trial.prefill);
            out.print("," + trial.nthreads);
            out.print("," + trial.nseconds);
            out.print("," + trial.trialno);
            out.print("," + trial.seed);
            //out.print("," + exp.startTime);
            //out.print("," + exp.endTime);
            double elapsedSeconds = (trial.endTime.get() - trial.startTime) / (double) NANOS_PER_SECOND;
            out.print("," + elapsedSeconds);
            out.print("," + trial.trueEnqueue);
            out.print("," + trial.trueDequeue);
            int throughput = (int) ((trial.trueEnqueue + trial.trueDequeue) / elapsedSeconds);
            out.print("," + throughput);
            out.print("\n");
        }
    }
    
    class Trial {
        AtomicLong endTime = new AtomicLong();
        volatile boolean stop = false;
        
        WorkThreadFactory fac;
        boolean prefill;
        int queueCapacity;
        int trialno;
        int nthreads;
        int[] threadProcBindings;
        double nseconds;
        int seed;
        Queue q;
        String qName;
        
        long startTime;
        //long endTime;
        int trueEnqueue;
        int trueDequeue;
        
        public Trial(WorkThreadFactory fac, boolean prefill, int queueCapacity, int trialno, double nseconds, int nthreads, int[] threadProcBindings, int seed, Queue q, String qName) {
            this.fac = fac;
            this.prefill = prefill;
            this.queueCapacity = queueCapacity;
            this.trialno = trialno;
            this.nseconds = nseconds;
            this.nthreads = nthreads;
            this.threadProcBindings = threadProcBindings;
            this.seed = seed;
            this.q = q;
            this.qName = qName;
        }
    }
    
    interface Operation {
        public void doOperation(Trial exp, WorkThread thread);
    }
    
    class RandomOperation implements Operation {
        public void doOperation(Trial exp, WorkThread thread) {
            if (thread.rng.nextInt() <= Integer.MAX_VALUE / 2) {
                if (exp.q.offer((long) thread.rng.nextInt(Integer.MAX_VALUE), thread.threadId, thread.rng)) thread.trueEnqueue++;
            } else {
                if (exp.q.poll(thread.threadId, thread.rng) != null) thread.trueDequeue++;
            }
        }
    }
    
    class ProducerOperation implements Operation {
        public void doOperation(Trial exp, WorkThread thread) {
            if (exp.q.offer((long) thread.rng.nextInt(Integer.MAX_VALUE), thread.threadId, thread.rng)) thread.trueEnqueue++;
        }
    }
    
    class ConsumerOperation implements Operation {
        public void doOperation(Trial exp, WorkThread thread) {
            if (exp.q.poll(thread.threadId, thread.rng) != null) thread.trueDequeue++;
        }
    }
    
    class WorkThread extends Thread {
        Operation op;
        final int INTERVAL = 50;
        CyclicBarrier start;
        Trial exp;
        Random rng;
        int threadId;
        int procBinding;
        
        int trueEnqueue = 0;
        int trueDequeue = 0;
        boolean selfTerminated = false;
        
        public WorkThread(Operation op, int procBinding, CyclicBarrier start, Trial exp, Random rng, int threadId) {
            this.op = op;
            this.procBinding = procBinding;
            this.start = start;
            this.exp = exp;
            this.rng = rng;
            this.threadId = threadId;
        }
        
        public void run() {
            if (procBinding != -1) {
                // TODO: DO THREAD BINDING TO VIRTUAL PROCESSOR HERE!!!
            }
            
            try {
                if (DEBUG) System.err.println("thread " + threadId + " awaiting");
                start.await();
                if (DEBUG) System.err.println("thread " + threadId + " passed barrier");
            } catch (Exception ex) {
                Logger.getLogger(Harness.class.getName()).log(Level.SEVERE, null, ex);
                System.exit(-1);
            }
            
            if (exp.startTime == 0) exp.startTime = System.nanoTime();
            
            int cnt = 0;
            while (!exp.stop && cnt < INTERVAL || System.nanoTime() < exp.startTime + exp.nseconds * NANOS_PER_SECOND) {
                if (cnt == INTERVAL) cnt = 0;
                else ++cnt;
                
                op.doOperation(exp, this);
            }
            
            // terminate the trial
            exp.stop = true;
            exp.endTime.compareAndSet(0, System.nanoTime());
            selfTerminated = true;
        }
    }
    
    private interface WorkThreadFactory {
        public WorkThread[] create(Trial exp, CyclicBarrier start);
    }
    
    // create workthreads to do random ops (enq/deq) with 50% probability each
    private class RandomWorkThreadFactory implements WorkThreadFactory {
        public WorkThread[] create(Trial exp, CyclicBarrier start) {
            assert(exp.nthreads > 0);
            
            java.util.Random rng = new java.util.Random(exp.seed);
            WorkThread[] threads = new WorkThread[exp.nthreads];
            for (int i=0;i<exp.nthreads;i++) {
                Operation op = new RandomOperation();
                threads[i] = new WorkThread(op,
                        (exp.threadProcBindings == null ? -1 : exp.threadProcBindings[i]),
                        start, exp, new Random(rng.nextInt()), i);
            }
            rng = null;
            return threads;
        }
    }
    
    // create workthreads for 50% producers, 50% consumers
    private class ProducerConsumerWorkThreadFactory implements WorkThreadFactory {
        public WorkThread[] create(Trial exp, CyclicBarrier start) {
            assert(exp.nthreads > 0);
            assert(exp.nthreads % 2 == 0);
            
            java.util.Random rng = new java.util.Random(exp.seed);
            WorkThread[] threads = new WorkThread[exp.nthreads];
            for (int i=0;i<exp.nthreads;i++) {
                Operation op = (i < exp.nthreads/2) ? new ProducerOperation() : new ConsumerOperation();
                threads[i] = new WorkThread(op,
                        (exp.threadProcBindings == null ? -1 : exp.threadProcBindings[i]),
                        start, exp, new Random(rng.nextInt()), i);
            }
            rng = null;
            return threads;
        }
    }
    
    void run(Trial trial) {
        
        // prefill data structure to 1/2 full
        
        if (trial.prefill) {
            for (int elt=0;elt<trial.queueCapacity/2;++elt) {
                final int threadId = 0;
                final Random rng = new Random(trial.seed);
                trial.q.offer((long) elt, threadId, rng);
            }
        }

        // create and prepare threads
        
        if (DEBUG) System.err.println("cyclic barrier has " + trial.nthreads + " slots");
        CyclicBarrier start = new CyclicBarrier(trial.nthreads);
        
        WorkThread[] threads = trial.fac.create(trial, start);
        
        // clean up before trial
        
        System.gc();

        // run trial
        
        for (int i=0;i<trial.nthreads;i++) {
            threads[i].start();
        }

        for (int i=0;i<trial.nthreads;i++) {
            try {
                if (DEBUG) System.err.println("waiting on thread " + i);
                threads[i].join((long) (trial.nseconds * 1000 * 1.5));
                break;
            } catch (InterruptedException ex) {
                Logger.getLogger(Harness.class.getName()).log(Level.SEVERE, null, ex);
                System.exit(-1);
            }
        }
        
        // kill any threads still running
        for (int i=0;i<trial.nthreads;i++) {
            if (!threads[i].selfTerminated) {
                threads[i].stop();
            }
        }
        
        // compile results
        
        trial.trueEnqueue = 0;
        trial.trueDequeue = 0;
        for (int i=0;i<trial.nthreads;i++) {
            trial.trueDequeue += threads[i].trueDequeue;
            trial.trueEnqueue += threads[i].trueEnqueue;
        }
        
        // clean up memory for next experiment
        
        trial.q = null;
        start = null;
        threads = null;
        System.gc();
    }
    
    public static void main(String[] args) {
        new Harness(System.out);
    }
}
